package xj.toolkit.netty.handler;

import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;

import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import xj.toolkit.ebus.EventBus;
import xj.toolkit.jmx.MBeanRegisterSupport;
import xj.toolkit.netty.transport.TransportUtils;
import xj.toolkit.netty.transport.Validateable;
import xj.toolkit.netty.transport.endpoint.Endpoint;
import xj.toolkit.netty.transport.endpoint.EndpointFactory;

import java.net.InetSocketAddress;
import java.util.concurrent.atomic.AtomicInteger;

@ChannelHandler.Sharable
public class TcpRequestHandler extends ChannelInboundHandlerAdapter {

	public static interface ChannelMXBean {

		public String getId();

		public boolean isConnected();

		public String getRemoteIp();

		public int getRemotePort();

		public String getLocalIp();

		public int getLocalPort();

		public String invokeRemoteMBeanOperation(final String objectName, final String operationName,
				final String[] params);
	}

	private static Logger logger = LoggerFactory.getLogger(TcpRequestHandler.class);

	private String eventRequestReceived;
	private String eventSessionOpened;
	private String eventSessionClosed;

	private EventBus eventBus;

	private EndpointFactory endpointFactory = null;

	private final MBeanRegisterSupport _mbeanSupport;

	private boolean registerSingleChannel = false;

	private AtomicInteger channelCount = new AtomicInteger(0);

	public TcpRequestHandler() {
		this("public:class=transport.tcp.netty.server,name=inbound");
	}

	public TcpRequestHandler(final String objectNamePrefix) {
		this._mbeanSupport = new MBeanRegisterSupport(objectNamePrefix, null);
	}

	public void destroy() {
		this._mbeanSupport.destroy();
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		logger.warn("channel:" + ctx.channel() + "/exceptionCaught: ", cause);
		ctx.channel().close();
	}

	@Override
	public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
		logger.debug("Message Received {}", msg);
		if (msg instanceof Validateable) {
			Object result = ((Validateable) msg).validate();
			if (result != null) {
				ctx.channel().writeAndFlush(result);
				logger.info("request {} validate failed,return resp {}", msg, result);
				return;
			}
		}
		Endpoint endpoint = TransportUtils.getEndpointOfSession(ctx);
		if (null != endpoint) {
			endpoint.messageReceived(ctx, msg);
		}

		if (eventBus != null && StringUtils.isNoneBlank(eventRequestReceived)) {
			eventBus.fireEvent(eventRequestReceived, msg);
		}
	}

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		if (logger.isInfoEnabled()) {
			logger.info("channelInactive: [{}]", ctx.channel().remoteAddress());
		}

		unregisterCtxMBean(ctx);

		if (StringUtils.isNoneBlank(eventSessionClosed) && eventBus != null) {
			eventBus.fireEvent(eventSessionClosed, ctx.channel());
		}

		Endpoint endpoint = TransportUtils.getEndpointOfSession(ctx);
		if (null != endpoint) {
			endpointFactory.endpointReleased(endpoint);
			endpoint.stop();
		}

		channelCount.decrementAndGet();
	}

	@Override
	public void channelActive(ChannelHandlerContext ctx) throws Exception {
		logger.info("channelActive:{}", ctx.channel());
		if (StringUtils.isNoneBlank(eventSessionOpened) && eventBus != null) {
			eventBus.fireEvent(eventSessionOpened, ctx.channel());
		}
		try {
			// create endpoint
			Endpoint endpoint = endpointFactory.createEndpoint(ctx.channel());
			if (null != endpoint) {
				TransportUtils.attachEndpointToSession(ctx, endpoint);
			}

			// register mbean for ctx
			registerCtxMBean(ctx);
		} catch (Exception ex) {
			logger.warn(" createEndpoint:", ex);
			ctx.channel().close();
		}

		channelCount.incrementAndGet();
	}

	private void registerCtxMBean(final ChannelHandlerContext ctx) {
		if (!registerSingleChannel) {
			return;
		}
		final Channel channel = ctx.channel();

		this._mbeanSupport.registerMBean(genChannelSuffix(channel), new ChannelMXBean() {

			@Override
			public String getId() {
				return channel.toString();
			}

			@Override
			public boolean isConnected() {
				return channel.isActive();
			}

			@Override
			public String getRemoteIp() {
				return ((InetSocketAddress) channel.remoteAddress()).getAddress().getHostAddress();
			}

			@Override
			public int getRemotePort() {
				return ((InetSocketAddress) channel.remoteAddress()).getPort();
			}

			@Override
			public String getLocalIp() {
				return ((InetSocketAddress) channel.localAddress()).getAddress().getHostAddress();
			}

			@Override
			public int getLocalPort() {
				return ((InetSocketAddress) channel.localAddress()).getPort();
			}

			@Override
			public String invokeRemoteMBeanOperation(final String objectName, final String operationName,
					final String[] params) {
				return null;
			}

		});

	}

	private void unregisterCtxMBean(final ChannelHandlerContext ctx) {

		if (!registerSingleChannel) {
			return;
		}

		final Channel channel = ctx.channel();

		this._mbeanSupport.unregisterMBean(genChannelSuffix(channel));
	}

	/**
	 * @param channel
	 * @return
	 */
	private static String genChannelSuffix(final Channel channel) {
		InetSocketAddress dstAddr = (InetSocketAddress) channel.remoteAddress();
		return "id=" + dstAddr.getAddress().getHostAddress() + "_" + dstAddr.getPort();
	}

	public String getEventRequestReceived() {
		return eventRequestReceived;
	}

	public void setEventRequestReceived(String eventRequestReceived) {
		this.eventRequestReceived = eventRequestReceived;
	}

	public String getEventSessionOpened() {
		return eventSessionOpened;
	}

	public void setEventSessionOpened(String eventSessionOpened) {
		this.eventSessionOpened = eventSessionOpened;
	}

	public String getEventSessionClosed() {
		return eventSessionClosed;
	}

	public void setEventSessionClosed(String eventSessionClosed) {
		this.eventSessionClosed = eventSessionClosed;
	}

	public void setEventBus(EventBus eventBus) {
		this.eventBus = eventBus;
	}

	public int getChannelCount() {
		return channelCount.get();
	}

	public void setEndpointFactory(EndpointFactory endpointFactory) {
		this.endpointFactory = endpointFactory;
	}

	public boolean isRegisterSingleChannel() {
		return registerSingleChannel;
	}

	public void setRegisterSingleChannel(boolean registerSingleChannel) {
		this.registerSingleChannel = registerSingleChannel;
	}

}
